package net.bwie.flink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.util.Collector;

/**
 * 基于Flink编写流式计算：交易订单事实统计，存储MySQL表中
 * @author xuanyu
 * @date 2025/10/15
 */
public class _02StreamMysqlSinkDemo {

	/**
	 * 交易订单：{"uid": "u101", "order_amount": 98.9}
	 */
	public static void main(String[] args) throws Exception{
		// 1-执行环境env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1) ;

		// 2-数据源source
		DataStreamSource<String> stream = env.socketTextStream("node101", 9999);

		// 3-数据转换transformation

		// 3-1.过滤脏数据
		SingleOutputStreamOperator<String> stream1 = stream.filter(
			new FilterFunction<String>() {
				@Override
				public boolean filter(String value) throws Exception {
					try{
						JSON.parseObject(value);
						return true;
					}catch (Exception e){
						return false;
					}
				}
			}
		);

		// 3-2.提取字段值
		SingleOutputStreamOperator<Tuple2<String, Double>> stream2 = stream1.process(
			new ProcessFunction<String, Tuple2<String, Double>>() {
				@Override
				public void processElement(String value, Context ctx, Collector<Tuple2<String, Double>> out) throws Exception {
					/**
					 * value 表示流中每条数据，ctx 表示上下文对象，进行很多高级操作，out表示数据输出
					 */
					// 解析json
					JSONObject jsonObject = JSON.parseObject(value);
					// 提起值
					Double orderAmount = jsonObject.getDouble("order_amount");
					// 二元组
					Tuple2<String, Double> tuple = Tuple2.of("amount", orderAmount);
					// 输出
					out.collect(tuple);
				}
			}
		);

		// 3-3. 分组
		KeyedStream<Tuple2<String, Double>, String> stream3 = stream2.keyBy(
			new KeySelector<Tuple2<String, Double>, String>() {
				@Override
				public String getKey(Tuple2<String, Double> value) throws Exception {
					return value.f0;
				}
			}
		);

		// 3-4. 求和
		SingleOutputStreamOperator<Tuple2<String, Double>> stream4 = stream3.sum(1);
		/*
			(amount, 10.0)
			(amount, 20.0)
			(amount, 30.0)
			(amount, 40.0)
		 */

		// 4-数据终端sink
//		stream4.print();
		/*
USE db_test ;
DROP TABLE IF EXISTS tbl_order_report;
CREATE TABLE IF NOT EXISTS tbl_order_report(
    order_report VARCHAR(255) PRIMARY KEY ,
    order_value DOUBLE
) ;

REPLACE INTO tbl_order_report(order_report, order_value) VALUES (?, ?) ;
		 */
		SinkFunction<Tuple2<String, Double>> jdbcSink = JdbcSink.sink(
			"REPLACE INTO tbl_order_report(order_report, order_value) VALUES (?, ?)",
			(statement, tuple) -> {
				statement.setObject(1, tuple.f0);
				statement.setObject(2, tuple.f1);
			},
			JdbcExecutionOptions.builder()
				.withBatchSize(1000)
				.withBatchIntervalMs(200)
				.withMaxRetries(5)
				.build(),
			new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
				.withUrl("jdbc:mysql://node101:3306/db_test")
				.withDriverName("com.mysql.jdbc.Driver")
				.withUsername("root")
				.withPassword("123456")
				.build()
		);
		stream4.addSink(jdbcSink);


		// 5-触发执行execute
		env.execute("FlinkStreamWordCount") ;
	}

}
